package jp.co.worksap.ate.gomoku.hadoopai;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import jp.co.worksap.ate.gomoku.localai.GoNode;
import jp.co.worksap.ate.gomoku.localai.GobanNode;
import jp.co.worksap.ate.gomoku.localai.GobanTree;
import jp.co.worksap.ate.gomoku.localai.LocalAIUtil;
import jp.co.worksap.ate.gomoku.localai.MultiLevelAI;
import jp.co.worksap.ate.gomoku.model.GoState;
import jp.co.worksap.ate.gomoku.model.Goban;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.IdentityReducer;
import org.apache.hadoop.mapreduce.Job;

public class GomokuMapReduce {
	public static class Map extends MapReduceBase implements
			Mapper<Text, GobanParam, Text, Text> {
		public void map(Text key, GobanParam value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			Goban goban = value.getGoban();
			GobanTree tree = new GobanTree(goban.clone(), value.getGoType(),
					value.getX(), value.getY());
			MultiLevelAI.computeGobanTree(tree, tree.getRoot(),
					value.getLevel(), value.getLevel());
			double score = tree.getSocre() - tree.getEnemyScore()
					+ tree.getRoot().getDefenceScore();
			int nodeNum = tree.getNodeNum();
			Text resultText = new Text("" + score + "_" + nodeNum);
			output.collect(key, resultText);
		}
	}

	public static class Reduce extends MapReduceBase implements
			Reducer<Text, Text, Text, Text> {
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}
	}

	public static java.util.Map<String, String> run(String folderName,
			Goban goban, GoState goType, List<GobanNode> rootList, int level)
			throws Exception {
		JobConf conf = HadoopUtil.getBasicConf();
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		conf.setMapperClass(Map.class);
		conf.setReducerClass(IdentityReducer.class);

		conf.setInputFormat(SequenceFileInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		Path outputPath = new Path("/gomoku/" + folderName + "/output");
		Path inputPath = new Path("/gomoku/" + folderName + "/input");

		conf.setJobName("gomoku-" + folderName);

		HadoopUtil.deleteGomokuFolder(folderName);
		HadoopUtil.createGomokuFolder(folderName);
		FileInputFormat.setInputPaths(conf, inputPath);
		FileOutputFormat.setOutputPath(conf, outputPath);
		HadoopUtil.writeInputFile(folderName, goban, goType, rootList, level);

		Job job = new Job(conf);
		job.waitForCompletion(true);
		java.util.Map<String, String> map = HadoopUtil
				.readOutputFile(folderName);
		HadoopUtil.deleteGomokuFolder(folderName);
		return map;
	}

	public static void main(String[] args) throws Exception {
		Goban goban = new Goban(19, 19);
		goban.putBlack(10, 10);
		goban.putBlack(10, 11);
		goban.putWhite(9, 10);
		List<GoNode> nodeList = LocalAIUtil
				.getAllValueableEmptyGoNode(goban, 2);
		List<GobanNode> rootList = new ArrayList<GobanNode>();
		GoState goType = GoState.WHITE;
		for (GoNode goNode : nodeList) {
			GobanNode rootNode = new GobanNode(goban.clone(), goType,
					goNode.getX(), goNode.getY());
			rootList.add(rootNode);
		}
		java.util.Map<String, String> map = run("test", goban, goType,
				rootList, 3);
		for (GobanNode goNode : rootList) {
			String key = goNode.getX() + "_" + goNode.getY();
			String score = map.get(key);
			System.out.println(goNode.getX() + " " + goNode.getY() + ":"
					+ score);
		}
	}
}
